package f5transformation.aggregation;

import flinkemp.Emp;
import flinkemp.EmpFun;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class KeyByTuple {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Emp> data = env.addSource(new EmpFun());
        data.print();

        SingleOutputStreamOperator<Tuple2<Integer, Double>> map = data.map(emp ->
                Tuple2.of(emp.deptNo, emp.sal)
        ).returns(new TypeHint<Tuple2<Integer, Double>>() {});

        KeyedStream<Tuple2<Integer, Double>, Integer> tuple = map.keyBy(t -> t.f0);
        tuple.sum(1).print("按部门编号求薪资和");
        tuple.min(1).print("按部门编号min");
        tuple.max(1).print("按部门编号max");
        tuple.minBy(1).print("按部门编号minBy");
        tuple.maxBy(1).print("按部门编号maxBy");
        env.execute();
    }
}